package a_aa_amainpackage.lvyou_clear.lv_update

import java.text.SimpleDateFormat
import java.util.Date

import a_aa_amainpackage.lvyou_clear.lv_update.util.updaue_addressfirst._
import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.functions.lit
import org.apache.spark.sql.{DataFrame, SQLContext, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}
import org.elasticsearch.spark._

import scala.collection.mutable.ArrayBuffer

/*
*项目名: o2odata_commend
*文件名: Tuniu
*创建者: hancongcong
*创建时间:2020/1/3 11:48
*描述: TODO
*/ object TuniuMp_yicahngzhengchang {
  val clust_obs = "s3a://"

  val data_base = "Tuniu"
  val collection = "tuniu_mp_detail_2207"

  val platform = "tuniu"
  val month = "7"
  val year = "2022"

  val time_stamp = "1659110400"
  val platform_Name = "tuniu_mp"


  //原始数据保存路径
  val source_obs_data_Path = clust_obs + s"o2o-tempdata/zsc/${year}/${month}/${data_base}/${platform_Name}"

  //价格库上月 "o2o-dataproces-group/hancongcong/tuniu_mp/handle_price_12"
  //  val pricePath = clust_obs + s"o2o-dataproces-group/zsc/platform/${platform_Name}/2021/handle_price_12"
  val pricePath = clust_obs + s"o2o-tempdata/zsc/platform/${platform_Name}/${year}/handle_price_${month.toInt - 1}"
  val newPricePath = clust_obs + s"o2o-tempdata/zsc/platform/${platform_Name}/${year}/handle_price_${month}"


  def main(args: Array[String]): Unit = {

    val sc = get_sparkContext()
    val sqlContext = new SQLContext(sc)
    sqlContext.udf.register("tranTimeToString", tranTimeToString _)

    /**
      * sql优化：
      * 1.考虑在where和order by涉及的列上建立索引，避免全表扫描
      * 2.应尽量避免在where子句中对字段进行null值判断，否则将导致引擎放弃使用索引而进行全表扫描，最好不要给数据库留null，可以在num值上设置默认值为0
      * 3.应尽量避免where子句中使用 != 或 <>操作符，负责将放弃使用索引而进行全表扫描
      * 4.应尽量避免再where子句中使用or来连接条件，如果一个字段有索引，一个字段没有索引，将导致引擎放弃使用索引而进行全表扫描(使用union all)
      * 5.in和not in也要慎用，否则会导致全表扫描，对于连续的数值，能用between就不要用in了，用exists代替in是一个好的选择
      * 6.在新建临时表时，如果一次性插入数据量很大，则可以使用select into
      */

    //旅游计算 销量，最后一个减第一个后平滑，价格历史取历史，最新取最新
    val ly_source_data = sqlContext.read.json(source_obs_data_Path)
    ly_source_data.show()

    //销量价格计算
    val priceData = tuniu_ticket(sqlContext.sparkSession, ly_source_data)
      .drop("salesAmount").withColumnRenamed("good_id", "product_Id")
    println(priceData.count())

    //对销量进行平滑
    val historySellData = hebingshuju(sc, sqlContext, priceData)
    val historySellRdd = fangchajisuan(sqlContext, historySellData)
    //当月销量与数据关联 与价格库关联
    val calcuData = guanlian(sqlContext, priceData, sqlContext.read.json(historySellRdd))._1

    // 打标签
    val sell_data = price_join(sqlContext, calcuData)._1.withColumnRenamed("product_Id", "good_id")
    sell_data.show()
    sell_data.where("cast(salesAmount as double) >0")
      .repartition(1).write.mode("overwrite").json(s"s3a://o2o-sourcedata-${year}/obs_result_data/${year}/${month}/${platform_Name}_sale")

    // 用于618期间计算数据
    //    sell_data.repartition(1).write.orc(clust_obs+s"o2o-dataproces-group/zyf/2021/618/${platform_Name}")

    val frame = updateaddress(sc, sqlContext, sqlContext.read.json(s"s3a://o2o-sourcedata-${year}/obs_result_data/${year}/${month}/${platform_Name}_sale"))
    //      .drop("add_to_field").where("cast(salesAmount as double)>0"))


    //地址再处理
    val frame1 = mp(frame, sc, sqlContext, year, month, "mp")
    frame1.registerTempTable("resultData")

    sqlContext.sql(
      """
        |select count(*),sum(sellCount),sum(salesAmount) from resultData
        |""".stripMargin).show(false)
    frame1.dropDuplicates("good_id").repartition(1).write.mode("overwrite").json(clust_obs + s"o2o-tempdata/shi_xinwang/tourism/${year}/${month}/${platform_Name}")

    sc.stop()
  }

  //拉数
  def get_sparkConf(): SparkConf = {
    val conf = new SparkConf()
    conf.setAppName(s"${this.getClass.getSimpleName}")
    conf.set("spark.debug.maxToStringFields", "500")
    conf.setMaster("local[*]")
    conf.set("es.nodes", "192.168.1.29")
    conf.set("es.port", "9200")
    conf.set("cluster.name", "O2OElastic")
    conf.set("spark.mongodb.input.uri", "mongodb://root:O2Odata123!@ 192.168.0.149:27017/admin") //  mongodb://ob:O2Odata123!@ 192.168.0.56:27017/admin   args(0)
    conf.set("spark.mongodb.input.database", s"${data_base}") //  库  args(1)
    conf.set("spark.mongodb.input.collection", s"${collection}") // 集合 args(2)
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    conf.set("spark.sql.caseSensitive", "true")
    conf.set("fs.s3a.access.key", "GAO7EO9FWKPJ8WFCQDME")
    conf.set("fs.s3a.secret.key", "LZ0xaHBSYKHaJ9ECDbX9f7zin79UZkXfGoNapRPL")
    conf.set("fs.s3a.endpoint", "https://obs.cn-north-1.myhuaweicloud.com")


    conf
  }


  def get_sparkContext(): SparkContext = {
    val conf = get_sparkConf()
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")
    sc
  }


  //销量和价格计算

  def tuniu_ticket(spark: SparkSession, dataFrame: DataFrame): DataFrame = {
    //    val data_Url = //dataValue

    //    val data_cacu = dataFrame.withColumnRenamed("sight_id","good_id").selectExpr("good_id", "add_to_field").toJSON.rdd.map(line => {
    val data_cacu = dataFrame.withColumnRenamed("sight_id", "good_id").selectExpr("good_id", "add_to_field")
      .toJSON.rdd.map(line => {

      // 将一行的数据转换成JSON对象
      val nObject = JSON.parseObject(line)

      // 数据转换为JSONARRAY，方便获取数据
      val add_to_field = nObject.getJSONArray("add_to_field")
      if (!add_to_field.isEmpty) {

        val priceText_temp_array = new ArrayBuffer[Double]
        val sell_temp = new ArrayBuffer[Long]
        var priceText = 0D

        var price_add = 0.0

        // 遍历JSON数组
        for (i <- 0 to add_to_field.size() - 1) {

          // 获取到对应的array转为对象
          val add_to_field_i: JSONObject = JSON.parseObject(add_to_field.get(i).toString)

          val price_add0 = add_to_field_i.getOrDefault("ticket_price", "-1").toString

          // 判断是否有子字符串 ~
          if (price_add0.contains("~")) price_add = (price_add0.split("~")(0).toDouble + price_add0.split("~")(1).toDouble) / 2
          if (!price_add0.contains("~")) price_add = price_add0.toDouble
          if (price_add > 0) priceText_temp_array += price_add

          val sell_add = add_to_field_i.getOrDefault("sold_count", "-1").toString.toLong
          if (i == 0) sell_temp += sell_add
          if (sell_add > 0) sell_temp += sell_add
        }
        if (priceText_temp_array.length >= 1) {
          //价格数组中的第一个价格
          val first_Price = priceText_temp_array(0)
          //价格数组中最后一个价格
          val last_Price = priceText_temp_array.last

          //价格最终值
          if (first_Price / last_Price <= 1.20 && first_Price / last_Price >= 0.80) {
            priceText = last_Price.formatted("%.2f").toDouble
          } else {
            priceText = first_Price.formatted("%.2f").toDouble
          }
        } else {
          priceText = 0
        }

        val sellCount: Long = if (sell_temp.length >= 1) sell_temp.last - sell_temp(0) else 0

        nObject.put("priceText", priceText)
        nObject.put("sellCount", sellCount)
        nObject.put("salesAmount", (priceText * sellCount).formatted("%.2f").toDouble)

      }
      nObject.toString

    })

    val data_cacuDF: DataFrame = spark.read.json(data_cacu)
    data_cacuDF.filter("sellCount>0").select("good_id", "sellCount", "salesAmount")
      .groupBy().agg("good_id" -> "count", "sellCount" -> "sum", "salesAmount" -> "sum").show()

    val dataDF = dataFrame.withColumnRenamed("sight_id", "good_id")
      .drop("add_to_field").join(data_cacuDF, Seq("good_id"), "left")
    dataDF


  }


  def hebingshuju(sc: SparkContext, sqlContext: SQLContext, priceData: DataFrame): DataFrame = {

    /* val esData = sc.esJsonRDD("2018_tuniu_mp,2019_tuniu_mp",
       s"""
          |{
          |  "size": 0
          |  , "query": {
          |    "match_all": {}
          |  }
          |}
           """.stripMargin).values
     val esSellData = sqlContext.read.json(esData)
       .withColumnRenamed("good_id", "product_Id")
       .select("product_Id", "sellCount")
       .unionAll(priceData.select("product_Id", "sellCount"))*/
    priceData
  }

  def fangchajisuan(sqlContext: SQLContext, source_sell: DataFrame): RDD[String] = {
    source_sell.registerTempTable("source_sell")
    val good_list = sqlContext.sql(
      """
        |select
        |product_Id,
        |collect_list(sellCount) over(partition by product_Id) as sell_time_list
        |from
        |source_sell
      """.stripMargin).dropDuplicates("product_Id")
    val res = good_list.toJSON.rdd.map(line => {
      val nObject = JSON.parseObject(line)
      val good_list = nObject.getOrDefault("sell_time_list", "-1").toString
      var array = JSON.parseArray(good_list)
      //数组总和
      var sumi = 0.0;
      for (i <- 0 to array.size() - 1) {
        sumi += array.get(i).toString.toDouble
      }
      //数组长度
      val array_size = array.size()
      //平均值
      var avg = (sumi / array_size).toDouble
      //方差里面 (x1-x)^2 +(x2-x)^2 +......(xn-x)^2)
      var sumfangcha = 0.toDouble
      for (i <- 0 to array.size() - 1) {
        sumfangcha += (array.get(i).toString.toInt - avg) * (array.get(i).toString.toInt - avg).toDouble
      }
      //方差 方差里面/数组长度
      val fangcha_res = sumfangcha / array_size
      //标准差
      val d = Math.sqrt(fangcha_res)
      val sell_newlist = new java.util.ArrayList[Int]()
      //判断数组中是否有大于等于标准差
      for (i <- 0 to array.size() - 1) {
        if (array.get(i).toString.toDouble >= d) {
          sell_newlist.add(array.get(i).toString.toInt)
        }
      }
      //新数组放小于方差de
      val ave_list = new java.util.ArrayList[Int]()
      for (i <- 0 to array.size() - 1) {
        if (array.get(i).toString.toDouble < d) {
          ave_list.add(array.get(i).toString.toInt)
        }
      }
      //求s数组的均值
      var size_other = ave_list.size()
      var sum_other = 0.0
      for (i <- 0 to ave_list.size() - 1) {
        sum_other += ave_list.get(i).toString.toDouble
      }
      //求去掉后的均值
      var avg_other = (sum_other / size_other).toDouble

      nObject.put("sell_newlist", sell_newlist)
      nObject.put("avg_cha", avg_other)
      nObject.put("sqrt_cha", d)
      nObject.toJSONString

    })
    res
  }

  def guanlian(sqlContext: SQLContext, source_9: DataFrame, srll_utils: DataFrame): (DataFrame, DataFrame) = {


    //销量库
    val value = srll_utils.toJSON.rdd.map(line => {
      val nObject = JSON.parseObject(line)
      val avg_cha = nObject.getOrDefault("avg_cha", "-1").toString
      nObject.put("avg_cha", avg_cha)
      nObject.toJSONString
    })
    val srll_util = sqlContext.read.json(value).where("avg_cha != '-1' ")
    println("srll_util:")
    // registerTempTable -- 注册临时表
    srll_util.registerTempTable("srll_utils")


    val dataDF = source_9
    dataDF.registerTempTable("source_9")

    // explode -- 炸裂函数，即->列转行
    val data1 = sqlContext.sql(
      """
        |select
        |t1.*,
        |explode(t2.sell_newlist) as sell_da,
        |t2.avg_cha
        |from
        |source_9 t1
        |join
        |srll_utils t2
        |on t1.product_Id=t2.product_Id
      """.stripMargin)

    data1.registerTempTable("data1")

    //销量大于标准差的
    val data2 = sqlContext.sql(
      """
        |select
        |*,
        |cast(avg_cha as bigint) as sell
        |from
        |data1
        |where sellCount=sell_da
      """.stripMargin).dropDuplicates("product_Id")
    data2.registerTempTable("data2")
    println("data2:")

    val data3 = sqlContext.sql(
      """
        |select
        |t1.*,
        |case when t2.product_Id is null then cast(t1.sellCount as bigint) else cast(t2.sell as bigint) end as sells
        |from
        |source_9 t1
        |left join
        |data2 t2
        |on t1.product_Id=t2.product_Id
      """.stripMargin).dropDuplicates("product_Id")
      .drop("sellCount", "salesAmount")
      .withColumnRenamed("sells", "sellCount")
    data3.registerTempTable("data3")

    println(data3.count())
    data3.show()

    //与价格库关联
    val priceData = sqlContext.read.json(pricePath)
    priceData.registerTempTable("priceData")
    val joinPriceData = sqlContext.sql(
      """
        |select
        |*,
        |(price*sellCount) as salesAmount
        |from
        |(select
        |t1.*,
        |case when t2.product_Id is null then t1.priceText else t2.priceText end as price
        |from
        |data3 t1
        |left join
        |priceData t2
        |on t1.product_Id=t2.product_Id)
      """.stripMargin).drop("priceText")
      .withColumnRenamed("price", "priceText")
      .dropDuplicates("product_Id")

    joinPriceData.registerTempTable("joinPriceData")
    joinPriceData.show()
    //生成这个月的价格库
    joinPriceData.selectExpr("product_Id", "priceText", s"tranTimeToString(cast(${time_stamp} as String)) as timeStamp")
      .union(priceData.selectExpr("product_Id", "priceText", s"${time_stamp}"))
      .repartition(1).write.mode("overwrite").json(newPricePath)

    val sum_res = sqlContext.sql(
      """
        |select
        |sum(sellCount) as sum_sell,
        |count(*) as count_good,
        |sum(salesAmount) as sum_sale
        |from
        |joinPriceData
      """.stripMargin)
    println("sum_res:")

    //    sum_res.show()
    (joinPriceData, sum_res)
  }


  def price_join(sqlContext: SQLContext, data3: DataFrame): (DataFrame, DataFrame) = {
    val dataDF = data3
      .withColumnRenamed("product_url", "goodUrl")
      .withColumnRenamed("sight_title", "title")
      .withColumnRenamed("sight_id", "good_id")
      .withColumnRenamed("pic_url", "images")
      .withColumnRenamed("sight_location", "address")
      .withColumnRenamed("good_count", "goodCount")
      .withColumnRenamed("general_count", "generalCount")
      .withColumnRenamed("poor_count", "poorCount")
      .withColumnRenamed("comment_score", "goodRatePercentage")
      .withColumnRenamed("comment_count", "commentCount")
      .withColumnRenamed("country", "countryAddress")
      .withColumn("platformName", lit("途牛门票"))
      .withColumn("timeStamp", lit(s"${time_stamp}"))
      .withColumn("platformName_spelling", lit(s"${platform_Name}"))
      .withColumn("platformId", lit("88"))
      .withColumn("shopType", lit("mp"))
      .withColumn("category_tag", lit("其他"))

    dataDF.show()
    (dataDF, dataDF)
  }

  def tranTimeToString(timestamp: String): String = {
    val tm = timestamp + "000"
    val fm = new SimpleDateFormat("yyyy-MM-dd")
    val time = fm.format(new Date(tm.toLong))
    time
  }


  def toes(data: DataFrame): Unit = {
    val value = data.toJSON.rdd.map(line => {
      val lines = JSON.parseObject(line)
      lines
    })
    value.saveToEs(s"${year}_update_${platform}_mp/${platform}_mp_${year}_${month}", Map("es.mapping.id" -> "good_id"))
  }

  def updateaddress(sc: SparkContext, sqlContext: SQLContext, frame: DataFrame): DataFrame = {

    //        val data = sc.esJsonRDD(s"${year}_update_tuniu_mp/tuniu_mp_${year}_${month}").values
    //        val frame = sqlContext.read.json(data)
    //        frame.registerTempTable("tuniu_mp")
    //        frame.repartition(1).write.json(s"s3a://o2o-dataproces-group/hancongcong/tuniu_mp/update/${year}/${month}")
    val value1 = frame.toJSON.rdd.map(line => {
      val nObject = JSON.parseObject(line)
      val good_id = nObject.getOrDefault("good_id", "-1").toString
      val address = nObject.getOrDefault("address", "-1").toString
      val administrative_region = nObject.getOrDefault("administrative_region", "-1").toString
      val economic_division = nObject.getOrDefault("economic_division", "-1").toString
      val if_state_level_new_areas = nObject.getOrDefault("if_state_level_new_areas", "-1").toString
      val poor_counties = nObject.getOrDefault("poor_counties", "-1").toString
      val urban_agglomerations = nObject.getOrDefault("urban_agglomerations", "-1").toString
      val city = nObject.getOrDefault("city", "-1").toString
      val city_grade = nObject.getOrDefault("city_grade", "-1").toString
      val city_origin = nObject.getOrDefault("city_origin", "-1").toString
      val district = nObject.getOrDefault("district", "-1").toString
      val district_origin = nObject.getOrDefault("district_origin", "-1").toString
      val if_city = nObject.getOrDefault("if_city", "-1").toString
      val if_district = nObject.getOrDefault("if_district", "-1").toString
      val province = nObject.getOrDefault("province", "-1").toString
      val regional_ID = nObject.getOrDefault("regional_ID", "-1").toString
      val the_belt_and_road_city = nObject.getOrDefault("the_belt_and_road_city", "-1").toString
      val the_belt_and_road_province = nObject.getOrDefault("the_belt_and_road_province", "-1").toString
      val the_yangtze_river_economic_zone_city = nObject.getOrDefault("the_yangtze_river_economic_zone_city", "-1").toString
      val the_yangtze_river_economic_zone_province = nObject.getOrDefault("the_yangtze_river_economic_zone_province", "-1").toString
      nObject.put("good_id", good_id)
      nObject.put("the_belt_and_road_city", the_belt_and_road_city)
      nObject.put("the_belt_and_road_province", the_belt_and_road_province)
      nObject.put("the_yangtze_river_economic_zone_city", the_yangtze_river_economic_zone_city)
      nObject.put("the_yangtze_river_economic_zone_province", the_yangtze_river_economic_zone_province)
      nObject.put("address", address)
      nObject.put("administrative_region", administrative_region)
      nObject.put("economic_division", economic_division)
      nObject.put("if_state_level_new_areas", if_state_level_new_areas)
      nObject.put("urban_agglomerations", urban_agglomerations)
      nObject.put("poor_counties", poor_counties)
      nObject.put("city", city)
      nObject.put("city_grade", city_grade)
      nObject.put("city_origin", city_origin)
      nObject.put("district", district)
      nObject.put("district_origin", district_origin)
      nObject.put("if_city", if_city)
      nObject.put("if_district", if_district)
      nObject.put("province", province)
      nObject.put("regional_ID", regional_ID)
      nObject.toJSONString
    })
    sqlContext.read.json(value1).registerTempTable("tuniu_mp")


    //地址
    val address_t = sqlContext.read
      .json(s"s3a://o2o-dimension-table/address_table/address_table_2020/${month}/address_platform/tuniu_ticket_address_2020_${month}/")
    address_t.registerTempTable("address_t")


    val datas1 = sqlContext.sql(
      """
        |select
        |a.*,
        |case when b.shopId is null then 0	else b.	address	end	address_xg	,
        |case when b.shopId is null then 0	else b.	administrative_region	end	administrative_region_xg	,
        |case when b.shopId is null then 0 else b.	aedzId	end	aedzId_xg	,
        |case when b.shopId is null then 0	else b.	city	end	city_xg	,
        |case when b.shopId is null then 0	else b.	city_grade	end	city_grade_xg	,
        |case when b.shopId is null then 0	else b.	city_origin	end	city_origin_xg	,
        |case when b.shopId is null then 0	else b.	district	end	district_xg	,
        |case when b.shopId is null then 0	else b.	district_origin	end	district_origin_xg	,
        |case when b.shopId is null then 0	else b.	economic_division	end	economic_division_xg	,
        |case when b.shopId is null then 0	else b.	if_city	end	if_city_xg	,
        |case when b.shopId is null then 0	else b.	if_district	end	if_district_xg	,
        |case when b.shopId is null then 0	else b.	if_state_level_new_areas	end	if_state_level_new_areas_xg	,
        |case when b.shopId is null then 0	else b.	latitude	end	latitude_xg	,
        |case when b.shopId is null then 0	else b.	longitude	end	longitude_xg	,
        |case when b.shopId is null then 0	else b.	name	end	name_xg	,
        |case when b.shopId is null then 0	else b.	poor_counties	end	poor_counties_xg	,
        |case when b.shopId is null then 0	else b.	province	end	province_xg	,
        |case when b.shopId is null then 0	else b.	regional_ID	end	regional_ID_xg	,
        |case when b.shopId is null then 0	else b.	registration_institution	end	registration_institution_xg	,
        |case when b.shopId is null then 0 	else b.	rural_demonstration_counties	end	rural_demonstration_counties_xg	,
        |case when b.shopId is null then 0	else b.	rural_ecommerce	end	rural_ecommerce_xg	,
        |case when b.shopId is null then 0	else b.	the_belt_and_road_city	end	the_belt_and_road_city_xg	,
        |case when b.shopId is null then 0	else b.	the_belt_and_road_province	end	the_belt_and_road_province_xg	,
        |case when b.shopId is null then 0	else b.	the_yangtze_river_economic_zone_city	end	the_yangtze_river_economic_zone_city_xg	,
        |case when b.shopId is null then 0	else b.	the_yangtze_river_economic_zone_province	end	the_yangtze_river_economic_zone_province_xg	,
        |case when b.shopId is null then 0	else b.	town	end	town_xg	,
        |case when b.shopId is null then 0	else b.	urban_agglomerations	end	urban_agglomerations_xg
        |from
        |tuniu_mp a
        |left join
        |address_t b
        |on a.good_id = b.shopId
          """.stripMargin).dropDuplicates("good_id")
      .drop("address")
      .drop("administrative_region")
      .drop("aedzId")
      .drop("city")
      .drop("city_grade")
      .drop("city_origin")
      .drop("district")
      .drop("district_origin")
      .drop("economic_division")
      .drop("if_city")
      .drop("if_district")
      .drop("if_state_level_new_areas")
      .drop("latitude")
      .drop("longitude")
      .drop("name")
      .drop("poor_counties")
      .drop("province")
      .drop("regional_ID")
      .drop("registration_institution")
      .drop("rural_demonstration_counties")
      .drop("rural_ecommerce")
      .drop("shopId")
      .drop("the_belt_and_road_city")
      .drop("the_belt_and_road_province")
      .drop("the_yangtze_river_economic_zone_city")
      .drop("the_yangtze_river_economic_zone_province")
      .drop("town")
      .drop("urban_agglomerations")


      .withColumnRenamed("address_xg", "address")
      .withColumnRenamed("administrative_region_xg", "administrative_region")
      .withColumnRenamed("aedzId_xg", "aedzId")
      .withColumnRenamed("city_xg", "city")
      .withColumnRenamed("city_grade_xg", "city_grade")
      .withColumnRenamed("city_origin_xg", "city_origin")
      .withColumnRenamed("district_xg", "district")
      .withColumnRenamed("district_origin_xg", "district_origin")
      .withColumnRenamed("economic_division_xg", "economic_division")
      .withColumnRenamed("if_city_xg", "if_city")
      .withColumnRenamed("if_district_xg", "if_district")
      .withColumnRenamed("if_state_level_new_areas_xg", "if_state_level_new_areas")
      .withColumnRenamed("latitude_xg", "latitude")
      .withColumnRenamed("longitude_xg", "longitude")
      .withColumnRenamed("name_xg", "name")
      .withColumnRenamed("poor_counties_xg", "poor_counties")
      .withColumnRenamed("province_xg", "province")
      .withColumnRenamed("regional_ID_xg", "regional_ID")
      .withColumnRenamed("registration_institution_xg", "registration_institution")
      .withColumnRenamed("rural_demonstration_counties_xg", "rural_demonstration_counties")
      .withColumnRenamed("rural_ecommerce_xg", "rural_ecommerce")
      .withColumnRenamed("shopId_xg", "shopId")
      .withColumnRenamed("the_belt_and_road_city_xg", "the_belt_and_road_city")
      .withColumnRenamed("the_belt_and_road_province_xg", "the_belt_and_road_province")
      .withColumnRenamed("the_yangtze_river_economic_zone_city_xg", "the_yangtze_river_economic_zone_city")
      .withColumnRenamed("the_yangtze_river_economic_zone_province_xg", "the_yangtze_river_economic_zone_province")
      .withColumnRenamed("town_xg", "town")
      .withColumnRenamed("urban_agglomerations_xg", "urban_agglomerations")


    val value = datas1.toJSON.rdd.map(line => {
      val nObject: JSONObject = JSON.parseObject(line)

      val Continent = nObject.getOrDefault("Continent", "-1").toString
      val Country = nObject.getOrDefault("Country", "-1").toString
      val City = nObject.getOrDefault("City", "-1").toString
      val District = nObject.getOrDefault("District", "-1").toString
      val Province = nObject.getOrDefault("Province", "-1").toString
      val SightZone = nObject.getOrDefault("SightZone", "-1").toString

      val destination_Province = nObject.getOrDefault("province", "-1").toString
      val destination_City = nObject.getOrDefault("city", "-1").toString
      val destination_District = nObject.getOrDefault("district", "-1").toString


      val obj = new JSONObject()
      obj.put("Continent", Continent)
      obj.put("Country", Country)
      obj.put("City", destination_City)

      nObject.put("city_out", destination_City)
      obj.put("District", destination_District)
      nObject.put("district_out", destination_District)
      obj.put("Province", destination_Province)
      nObject.put("province_out", destination_Province)

      obj.put("SightZone", SightZone)
      nObject.put("destination", obj)
      nObject.toString
    })



    //        val value2 = sqlContext.read.json(value).where("regional_ID='0' ")
    //        println(value2.count())
    //        value2.select("good_id").show(500)


    sqlContext.read.json(value)


  }


}
